package cn.getech.data.metadata.handler;

import cn.getech.data.development.entity.ConfConnect;
import cn.getech.data.development.service.ConfConnectService;
import cn.getech.data.development.utils.SecurityUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.ExceptionUtils;
import cn.getech.data.metadata.constant.ConnectTypeEnum;
import cn.getech.data.metadata.constant.TaskStatusEnum;
import cn.getech.data.metadata.db.*;
import cn.getech.data.metadata.entity.meta.CollectionTask;
import cn.getech.data.metadata.service.CollectionTaskService;
import cn.getech.data.metadata.utils.MetadataContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.function.Consumer;

/**
 * 自动采集任务执行类
 */
@Slf4j
@Component("taskExecuteHandler")
public class TaskExecuteHandler {
    private ConfConnectService confConnectService;
    private SecurityUtil securityUtil;
    private CollectionTaskService collectionTaskService;

    private void updateTaskStatus(CollectionTask task, Integer runStatus){
        CollectionTask newTask = new CollectionTask();
        newTask.setId(task.getId());
        newTask.setRunStatus(runStatus);
        newTask.setRunTime(new Date());
        task.setRunTime(newTask.getRunTime());
        collectionTaskService.updateById(newTask);
    }

    private Thread thread;

    public void oneExecute(String param, Consumer<Boolean> complete) throws Exception {
        thread = new Thread(() -> {
            try {
                execute(param);
            } catch (Exception e) {
                log.error("oneExecute", e);
            } finally {
                complete.accept(true);
            }
        });
        thread.start();
    }

    public void oneStop() {
        if(thread != null){
            thread.stop();
        }
    }
    //必须公共方法  其他地方用反射使用到
    public void execute(String param) throws Exception {
        this.confConnectService = MetadataContextUtil.getBean(ConfConnectService.class);
        this.securityUtil = MetadataContextUtil.getBean(SecurityUtil.class);
        this.collectionTaskService = MetadataContextUtil.getBean(CollectionTaskService.class);

        CollectionTask newTask = new CollectionTask();
        CollectionTask task = null;
        long startTime = System.currentTimeMillis();
        try {
            long taskId = Long.parseLong(param);
            task = collectionTaskService.getById(taskId);
            if (task == null)
                throw new RRException("没有获取到采集任务 id:" + taskId);

            updateTaskStatus(task, TaskStatusEnum.RUNNING.getValue());

            execute(task, startTime);

            newTask.setRunStatus(TaskStatusEnum.SUCCESS.getValue());
        } catch (Exception e) {
            if(task != null){
                try{
                    //添加错误日志
                    task.setRunStatus(TaskStatusEnum.FAIL.getValue());
                    String error = ExceptionUtils.getExceptionDetail(e);
                    collectionTaskService.insertRunLog(startTime, task, error, task.getTaskRunLeader());
                    newTask.setRunStatus(TaskStatusEnum.FAIL.getValue());
                }catch (Exception e2){
                    throw e2;
                }
            }
            throw e;
        } finally {
            if(task != null){
                //更改运行次数
                newTask.setId(task.getId());
                newTask.setRunCount(task.getRunCount() == null ? 1 : task.getRunCount() + 1);
                collectionTaskService.updateById(newTask);
            }
        }
    }

    private void execute(CollectionTask task, long startTime) throws Exception {
        if (task.getSourceId() == null)
            throw new RRException("采集任务没有配置数据源 任务名称:" + task.getTaskName());

        ConfConnect confConnect = confConnectService.getById(task.getSourceId());
        if (confConnect == null)
            throw new RRException("没有获取到数据源 任务名称:" + task.getTaskName());

        confConnect.setPassword(securityUtil.decrypt(confConnect.getPassword()));

        ConnectTypeEnum t = ConnectTypeEnum.getDriver(confConnect.getTypeId());
        String url = "";
        if(StringUtils.isEmpty(confConnect.getDbname())){
            url = String.format(t.getUrl(), confConnect.getHost(), confConnect.getPort(), "");
        }else {
            url = String.format(t.getUrl(), confConnect.getHost(), confConnect.getPort(), confConnect.getDbname());
        }
        DBStructure dbStructure = null;
        if (ConnectTypeEnum.MySQL.getCode().equals(confConnect.getTypeId())) {
            dbStructure = new MysqlStructure(url, confConnect.getUsername(), confConnect.getPassword(),
                    ConnectTypeEnum.MySQL.getDriverClass(), confConnect.getDbname(), task);
        } else if (ConnectTypeEnum.Oracle.getCode().equals(confConnect.getTypeId())) {
            dbStructure = new OracleStructure(url, confConnect.getUsername(), confConnect.getPassword(),
                    ConnectTypeEnum.Oracle.getDriverClass(), confConnect.getDbname(), task);
        } else if (ConnectTypeEnum.SQL_Server.getCode().equals(confConnect.getTypeId())) {
            dbStructure = new SqlServerStructure(url, confConnect.getUsername(), confConnect.getPassword(),
                    ConnectTypeEnum.SQL_Server.getDriverClass(), confConnect.getDbname(), task);
        } else if (ConnectTypeEnum.PostgreSQL.getCode().equals(confConnect.getTypeId())) {
            dbStructure = new PostgreSQLStructure(url, confConnect.getUsername(), confConnect.getPassword(),
                    ConnectTypeEnum.PostgreSQL.getDriverClass(), confConnect.getDbname(), task);
        } else if (ConnectTypeEnum.Hive.getCode().equals(confConnect.getTypeId())) {
            dbStructure = new HIveStructure(confConnect,url, task);
        }
        if (dbStructure != null) {
            dbStructure.generateTable(startTime);
        } else {
            throw new RRException("不支持 " + t.getName() + " 类型数据源擦剂");
        }
    }
}
